package com.tn.rtdata.examples.stockticker.server.fill;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.tn.rtdata.DataEvent;
import com.tn.rtdata.DataEventStats;
import com.tn.rtdata.DataListener;
import com.tn.rtdata.Fill;
import com.tn.rtdata.FillException;
import com.tn.rtdata.FillIdentity;
import com.tn.rtdata.FillInfo;
import com.tn.rtdata.examples.stockticker.StockTick;
import com.tn.rtdata.util.ObservableMap;

/**
 * A specialization of <code>Fill</code> that handles <code>StockTick</code>s.
 */
public class StockTickFill extends Fill<String, StockTick>
{
  private static final Logger LOGGER = LoggerFactory.getLogger(StockTickFill.class);

  private Thread lockingThread;
  private int lockCount;
  private ObservableMap<String, StockTick> stockTickCache;
  private StockTickListener stockTickListener;

  /**
   * Creates a new <code>StockTickFill</code>.
   *
   * @param fillIdentity        the fill identity.
   * @param stockTickCache      the cache the stock ticks will be kept in.
   * @param fillInfoCache       the cache the fill info will be kept in.
   * @param dataEventStatsCache the cache the data event stats will be held in.
   */
  public StockTickFill(
    FillIdentity fillIdentity,
    ObservableMap<String, StockTick> stockTickCache,
    Map<FillIdentity, FillInfo<String, StockTick>> fillInfoCache,
    Map<String, DataEventStats> dataEventStatsCache
  )
  {
    super(fillIdentity, fillInfoCache, dataEventStatsCache);
    this.stockTickCache = stockTickCache;
    this.stockTickListener = new StockTickListener();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected String getIdentity(StockTick stockTick)
  {
    return stockTick.getStockId();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected boolean isLocked()
  {
    return lockingThread != null && !lockingThread.equals(Thread.currentThread());
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void afterStart()
  {
    stockTickCache.addDataListener(stockTickListener);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void beforeStop()
  {
    stockTickCache.removeDataListener(stockTickListener);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void lock()
  {
    //Block until the lock is available.
    while (isLocked())
    {
      try
      {
        Thread.sleep(50);
      }
      catch (InterruptedException e)
      {
        //Ignore.
      }
    }

    lockingThread = Thread.currentThread();
    lockCount++;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void unlock()
  {
    lockCount--;
    if (lockCount == 0)
    {
      lockingThread = null;
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected Collection<StockTick> values()
  {
    List<StockTick> stockTicks = new ArrayList<StockTick>();

    for (Object stockId : getFillIdentity().getParameters())
    {
      //noinspection SuspiciousMethodCalls
      stockTicks.add(stockTickCache.get(stockId));
    }

    return stockTicks;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void destroy()
  {
    stockTickCache.removeDataListener(stockTickListener);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected Map<String, StockTick> initializeFill() throws FillException
  {
    Map<String, StockTick> stockTicks = new HashMap<String, StockTick>();

    for (Object stockCode : getFillIdentity().getParameters())
    {
      StockTick stockTick = new StockTick(stockCode.toString());
      stockTicks.put(stockTick.getStockId(), stockTick);

      if (!stockTickCache.containsKey(stockTick.getStockId()))
      {
        stockTickCache.put(stockTick.getStockId(), stockTick);
      }
    }

    return stockTicks;
  }

  /**
   * An implementation of <code>DataListener</code> that fires fill events.
   */
  private class StockTickListener implements DataListener<StockTick>
  {
    /**
     * {@inheritDoc}
     */
    @Override
    public void dataEvent(DataEvent<StockTick> dataEvent)
    {
      switch (dataEvent.getType())
      {
        case ADD:
          if (getFillIdentity().getParameters().contains(dataEvent.getNewValue().getStockId()))
          {
            fireEntryAdded(dataEvent.getNewValue());
          }
          return;

        case UPDATE:
          if (getFillIdentity().getParameters().contains(dataEvent.getNewValue().getStockId()))
          {
            fireEntryUpdated(dataEvent.getNewValue(), dataEvent.getOldValue());
          }
          return;

        case DELETE:
          if (getFillIdentity().getParameters().contains(dataEvent.getOldValue().getStockId()))
          {
            fireEntryDeleted(dataEvent.getOldValue());
          }
          return;

        default:
          LOGGER.warn("Unrecognized data event type: {}", dataEvent.getType());
      }
    }
  }
}